package com.xahj.bd2104.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created with IntelliJ IDEA.
 * Author: Amos
 * E-mail: amos@amoscloud.com
 * Date: 2021/9/11
 * Time: 11:31
 * Description: 
 */
object SparkCoreDemo1_WordCount {
  def main(args: Array[String]): Unit = {
    // SparkContext是SparkCore的程序入口

    // 0. 创建Spark配置文件对象
    val conf = new SparkConf()
    // 0.1 添加集群的master节点url
    //  提交任务到集群时设置为集群的master节点url
    //  conf.setMaster("spark://node01:7077")
    //  本地测试时 设置为本地模式
    //  local 单线程
    //  local[n] 给当前任务分配n个cpu核心  n<=本机可用cpu核心数
    //  local[*] 给当前任务分配可用的最大核心数
    conf.setMaster("local[4]")

    // 0.2 设置spark程序的名称
    // 名称用于直观的区分当前任务的内容
    // 在MasterUI上可以直接对响应的任务进行终止操作
    conf.setAppName("job_WordCount")

    // 0.3 可以在SparkConf中添加其他需要的配置项
    //    conf.set("key","value")


    // 1. 创建SparkContext对象实例
    val sc = new SparkContext(conf)


    // 2. sc读取文本文件创建RDD

    val path = "C:\\Users\\Amos\\Desktop\\Harry.txt"
    val rdd: RDD[String] = sc.textFile(path)

    // 3. 将每一行数据切分成单词
    val rdd1: RDD[String] = rdd.flatMap(_.split("\\s+"))

    // 4. 清洗 : 所有单词转小写并去除非单词字符 转换成二元组
    val rdd2: RDD[(String, Int)] = rdd1.map(w => {
      val word = w.toLowerCase
        .replaceAll("\\W", "")
      (word, 1)
    })

    // 5. 按照Key进行聚合
    val rdd3: RDD[(String, Int)] = rdd2.reduceByKey(_ + _)

    // 6. 按照数量进行降序排列

    val rdd4: RDD[(String, Int)] = rdd3.sortBy(_._2, false)

    // 7. 输出

    rdd4
      .repartition(1)
      .foreach(println)

  }
}
